package com.kuiniu.threeframe.client;

import com.kuiniu.threeframe.Address;
import com.kuiniu.threeframe.client.entity.Message;
import com.kuiniu.threeframe.client.timeout.MsgTimeoutManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Client {

    private int connectTimeout;
    private int readTimeout;
    private int writeTimeout;
    private Address address;
    private SocketChannel socketChannel;
    private ClientState state = ClientState.closed;
    private ExecutorService service;
    private Bootstrap bootstrap;
    private ChannelInitializer<Channel> initializer;
    private int resendCount;
    private int resendInterval;
    private MsgTimeoutManager msgTimeoutManager;
    private List<EventListener> msgListeners;
    private volatile boolean manualClosed = false;

    public enum ClientState {
        closed, connecting, connected, failed
    }

    public Client(Builder builder) {
        connectTimeout = builder.connectTimeout;
        readTimeout = builder.readTimeout;
        writeTimeout = builder.writeTimeout;
        address = builder.address;
        resendCount = builder.resendCount;
        msgListeners = new ArrayList<>();
        resendInterval = builder.resendInterval;
        msgTimeoutManager = new MsgTimeoutManager(this);
        initializer = new DefaultChannelInitializer(this);
        service = Executors.newSingleThreadExecutor(r -> new Thread(r, "Netty-client-thread"));
        newConnect();
    }

    public int getConnectTimeout() {
        return connectTimeout;
    }

    public int getReadTimeout() {
        return readTimeout;
    }

    public int getWriteTimeout() {
        return writeTimeout;
    }

    public Address getAddress() {
        return address;
    }

    public int getResendCount() {
        return resendCount;
    }

    public int getResendInterval() {
        return resendInterval;
    }

    public MsgTimeoutManager getMsgTimeoutManager() {
        return msgTimeoutManager;
    }

    public void addListener(EventListener listener) {
        msgListeners.add(listener);
    }

    public List<EventListener> getMsgListeners() {
        return msgListeners;
    }

    public static final class Builder {
        int connectTimeout;
        int readTimeout;
        int writeTimeout;
        Address address;
        int resendCount;
        int resendInterval;

        public Builder() {
            connectTimeout = 5 * 1000;
            readTimeout = 60 * 1000;
            writeTimeout = 60 * 1000;
            resendCount = 5;
            resendInterval = 8 * 1000;
        }

        public Builder connectTimeout(int connectTimeout) {
            this.connectTimeout = connectTimeout;
            return this;
        }

        public Builder readTimeout(int readTimeout) {
            this.readTimeout = readTimeout;
            return this;
        }

        public Builder writeTimeout(int writeTimeout) {
            this.writeTimeout = writeTimeout;
            return this;
        }

        public Builder address(Address address) {
            this.address = address;
            return this;
        }

        public Builder resendCount(int resendCount) {
            this.resendCount = resendCount;
            return this;
        }

        public Builder resendInterval(int resendInterval) {
            this.resendInterval = resendInterval;
            return this;
        }

        public Client build() {
            return new Client(this);
        }
    }

    private void newConnect() {
        service.execute(this::connect);
    }


    public void connect() {
        System.out.println("开始向" + address.getUrl() + "发起socket连接请求");
        if (!manualClosed && state != ClientState.connecting) {
            try {
                if (bootstrap != null && bootstrap.group() != null) {
                    bootstrap.group().shutdownGracefully();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                bootstrap = null;
            }
            state = ClientState.connecting;
            initBootstrap();
            connectToServer();
        }
    }


    public void close() {
        System.out.println("主动断开连接");
        if (manualClosed) {
            return;
        }
        manualClosed = true;
        try {
            if (bootstrap != null && bootstrap.group() != null) {
                bootstrap.group().shutdownGracefully();
            }
            if (socketChannel != null) {
                socketChannel.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bootstrap = null;
            socketChannel = null;
            state = ClientState.closed;
            service.shutdown();
        }
    }

    public boolean isConnected() {
        return state == ClientState.connected;
    }

    private void initBootstrap() {
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(loopGroup).channel(NioSocketChannel.class);
        // 设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        // 设置禁用nagle算法
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        // 设置连接超时时长
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
        // 设置初始化Channel
        bootstrap.handler(initializer);
    }

    private void connectToServer() {
        ChannelFuture future;
        try {
            future = bootstrap.connect(address.getUrl(), address.getPort()).sync();
            if (future.isSuccess()) {
                socketChannel = (SocketChannel) future.channel();
                state = ClientState.connected;
            } else {
                state = ClientState.failed;
            }
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();

        } finally {
            if (!manualClosed) {
                service.execute(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(5);
                        state = ClientState.failed;
                        connect();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
        }
    }

    public void sendMsg(Message message, boolean resend) {
        if (message != null) {
            int len = message.getData().length();
            String sendStr = "##" +
                    String.format("%04d", len) +
                    message.getData();
            System.out.println("向" + address.getUrl() + "发送数据:" + sendStr);
            if (socketChannel != null) {
                socketChannel.writeAndFlush(sendStr);
                if (resend && msgTimeoutManager != null) {
                    msgTimeoutManager.addMsg(message);
                }
            }
        }
    }

}
